﻿using System;
using System.Threading;
using eLoop;

namespace eNets.CBK.Pipe;

abstract class APipe<T> : IRunnable
{
    public APipe(ITaskScheduler scheduler)
    {
        this.scheduler = scheduler;
    }

    private ITaskScheduler scheduler;
    private Action listener;
    private long wait = 0;

    public abstract long ReadableLength
    {
        get;
    }

    public abstract IPipeReader<T> Reader
    {
        get;
    }

    public abstract IPipeWriter<T> Writer
    {
        get;
    }
    
    public void AddListener(Action listener)
    {
        if (listener == null)
        {
            throw new ArgumentNullException(nameof(listener));
        }

        this.listener = listener;
    }
    
    public void RcvAsync()
    {
        Interlocked.Exchange(ref wait, 1);
        this.Notify();
    }

    public virtual void ReaderFlush()
    {
    }

    public void WriterFlush()
    {
        this.Notify();
    }

    private void Notify()
    {
        if (Interlocked.CompareExchange(ref this.wait, 0, 1) == 1)
        {
            if (this.ReadableLength > 0)
            {
                this.scheduler.Schedule(this, null);
            }
        }
    }
    
    void IRunnable.Execute(object state)
    {
        this.listener.Invoke();
    }
}